package com.atguigu.flink.datastramapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Created by Smexy on 2023/2/27
 *
 *  JdbcSink.sink(
 *       	sqlDmlStatement,                       // mandatory
 *       	jdbcStatementBuilder,                  // mandatory
 *       	jdbcExecutionOptions,                  // optional  可选
 *       	jdbcConnectionOptions                  // mandatory
 * );
 */
public class Demo6_JDBCSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
            String sql: 要使用的sql
            JdbcStatementBuilder<T> statementBuilder: 用于填充语句中的占位符.
                    T: 要写出的类型
            JdbcExecutionOptions executionOptions:  可选的执行参数，按照批次执行。多少条一个批次等等
            JdbcConnectionOptions connectionOptions: 如何连接数据库
         */
        SinkFunction<WaterSensor> sink = JdbcSink.sink(
            "insert into watersensor values(?,?,?)",
            new JdbcStatementBuilder<WaterSensor>()
            {
                /*
                    使用ps，将 一条 WaterSensor的数据，填入sql的占位符
                 */
                @Override
                public void accept(PreparedStatement ps, WaterSensor value) throws SQLException {
                    ps.setString(1, value.getId());
                    ps.setLong(2, value.getTs());
                    ps.setInt(3, value.getVc());

                }
            },
            JdbcExecutionOptions.builder()
                                .withBatchSize(100)  //一个批次多少条
                                .withMaxRetries(3)  //写失败了，最多重试多少次
                                .withBatchIntervalMs(10000) //当前数据不满一个批次，根据时间间隔写出
                                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:clickhouse://hadoop102:8123/upp220926")
                .build()
        );

        env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .addSink(sink);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
